package core

import (
	"container/list"
	"encoding/binary"
	"math"
	"sync"
)

//每个消息只能被消费一次
//消息状态：未消费，处理中，已消费
//轮询：按消费者顺序分发，竞争：消费者主动拉取，谁先请求谁获取，分发消息
//消费者处理结束后，需发ACK确认消息告知通道。
//每个消费者需挂靠任务数量，用于标识“忙闲”程度，方便分发。

const (
	Message_NoUse  = byte(0x00) //未消费
	Message_Useing = byte(0x01) //消费中
	Message_Finish = byte(0x02) //已消费
)

type OneUserMessage struct {
	Msg   *TopicMessage
	State byte
	//OverTimer *time.Timer //超时机制,如果该消息在指定时间内未使用完成，则收回消息，再次分配
}

func NewOneUserMessage(msg *TopicMessage) *OneUserMessage {
	return &OneUserMessage{Msg: msg, State: Message_NoUse}
}

// ///////////////////////////////////////////////
type Topic_OneUser struct {
	Mng map[*TopicFromDB]*OnlyOne

	ProcessEvent chan int
}

var (
	Topic_OneUser_instance *Topic_OneUser
	Topic_OneUser_once     sync.Once
)

func GetTopic_OneUser() *Topic_OneUser {
	Topic_OneUser_once.Do(func() {
		Topic_OneUser_instance = &Topic_OneUser{Mng: make(map[*TopicFromDB]*OnlyOne)}
	})
	return Topic_OneUser_instance
}
func (t *Topic_OneUser) ProcessTopic_OneUser(topic *TopicFromDB, msg *TopicMessage) {
	one := t.Select_OneUser(topic)
	one.AddMessage(msg)

	one.Tasks <- 1
}
func (t *Topic_OneUser) ProcessFinishAckMsg(data []byte) {
	if len(data) < 13 {
		return
	}

	if data[0] != MQ_FinishAck {
		return
	}
	msgid := binary.BigEndian.Uint64(data[1:9])

	name, err := ReadStringFromBytes(data[9:])
	if err != nil {
		return
	}
	obj := t.Select_OneUserFromName(name)
	if obj != nil {
		obj.FinishMessage(msgid)
		obj.CleanFinishMsg()
	}
}
func (t *Topic_OneUser) Select_OneUserFromName(topic string) *OnlyOne {
	var result *OnlyOne = nil
	for obj, one := range t.Mng {
		if obj.Name == topic {
			result = one
			break
		}
	}
	return result
}
func (t *Topic_OneUser) Select_OneUser(topic *TopicFromDB) *OnlyOne {
	var result *OnlyOne = nil
	if obj, err := t.Mng[topic]; !err {
		result = NewOnlyOne(topic)
		result.StartTask()
		t.Mng[topic] = result
		return result
	} else {
		return obj
	}
}

// //////////////////////////////////////////////////////
type OnlyOne struct {
	topic *TopicFromDB
	queue *list.List
	mutex sync.Mutex
	Tasks chan int
}

func NewOnlyOne(topic *TopicFromDB) *OnlyOne {
	return &OnlyOne{topic: topic, queue: list.New(), Tasks: make(chan int, 1000)}
}
func (c *OnlyOne) StartTask() {
	go func() {
		for range c.Tasks {
			for c.ScanAndProcess() {

			}
		}
	}()
}
func (c *OnlyOne) AddMessage(msg *TopicMessage) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	c.queue.PushBack(NewOneUserMessage(msg))
}
func (t *OnlyOne) ScanAndProcess() bool {
	msg1 := t.GetFirstNoUseMessage()
	if msg1 == nil {
		return false
	}
	client := t.SelectUser()
	if client != nil {
		client.MsgTaskNum++
		msg1.Msg.Retain = 0x01
		client.SendTopicMsg(msg1.Msg)
		return true
	}
	return false
}

// 选择一个合适的消费者
func (c *OnlyOne) SelectUser() *Client {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	if len(c.topic.Clients) == 0 {
		return nil
	}
	//只有一个，没得选，只能选他
	if len(c.topic.Clients) == 1 {
		for _, client := range c.topic.Clients {
			return client
		}
	}

	objs := NewList[*Client]()

	minCount := math.MaxUint32
	for _, client := range c.topic.Clients {
		if client.MsgTaskNum == uint32(minCount) {
			objs.Append(client)
			continue
		}
		if client.MsgTaskNum < uint32(minCount) {
			minCount = int(client.MsgTaskNum)
			objs.Clear()
			objs.Append(client)
			continue
		}
		// if client.MsgTaskNum == 0 {
		// 	return client
		// }
	}

	//只有唯一的一个
	if objs.Len() == 1 {
		if obj, err := objs.Get(0); err == nil {
			return obj
		}
	} else {
		//随机抽取
		nn := GetRand(0, objs.Len()-1)

		if client, err := objs.Get(nn); err == nil {
			return client
		}
	}
	return nil
}

// 该消息标志已消费完成，并关闭定时器
func (c *OnlyOne) FinishMessage(msgid uint64) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	if c.queue.Len() <= 0 {
		return
	}
	for e := c.queue.Front(); e != nil; e = e.Next() {
		msg := e.Value.(*OneUserMessage)
		if msg.Msg.MessageID == msgid {
			msg.State = Message_Finish
			//msg.OverTimer.Stop()
		}
	}
}

// 清理已消费完成的消息
func (c *OnlyOne) CleanFinishMsg() {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	if c.queue.Len() <= 0 {
		return
	}
	for e := c.queue.Front(); e != nil; e = e.Next() {
		msg := e.Value.(*OneUserMessage)
		if msg.State == Message_Finish {
			//msg.OverTimer.Stop()
			c.queue.Remove(e)
		}
	}
}

// 按顺序获取第一个未消费的消息,并将状态设置为已消费状态
func (c *OnlyOne) GetFirstNoUseMessage() *OneUserMessage {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	var result *OneUserMessage = nil

	if c.queue.Len() <= 0 {
		return nil
	}
	for e := c.queue.Front(); e != nil; e = e.Next() {
		msg := e.Value.(*OneUserMessage)
		if msg.State == Message_NoUse {
			msg.State = Message_Useing
			//msg.OverTimer = time.NewTimer(MQ_OverTime * time.Millisecond)
			result = msg
		}
	}
	// 超时，则将该消息状态设为未消费
	// if result != nil {
	// 	go func() {
	// 		<-result.OverTimer.C
	// 		//c.mutex.Lock()
	// 		result.State = Message_NoUse
	// 		//c.mutex.Unlock()
	// 		c.Tasks <- 1
	// 		fmt.Println("-------------------------------------------")
	// 	}()
	// }
	return result
}
